package com.evan.common.service.impl;

import com.evan.common.bit.CommandConstEnum;
import com.evan.common.bit.ZkServerModeEnum;
import com.evan.common.bit.ZkServerStatus;
import com.evan.common.dao.ZkServerDao;
import com.evan.common.exception.ZkException;
import com.evan.common.pojo.ZkServerInfo;
import com.evan.common.service.ZkServerService;
import com.evan.common.utils.SocketClient;
import com.evan.common.utils.ZookeeperClient;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
public class ZkServerServiceImpl implements ZkServerService {

    @Autowired
    private ZkServerDao zsd;

    @Override
    public List<ZkServerInfo> getAllZkNode() {
        return zsd.findAll();
    }

    @Override
    public ZkServerInfo addZkNode(ZkServerInfo zk) throws Exception {
        Map<String, String> hostMap = this.serverStatusByIpPort(zk.getIp(), zk.getPort());
        if(hostMap.get(zk.getIp()).equals("leader")){
           zk.setServerMode(ZkServerModeEnum.LEADER);
        }else{
            zk.setServerMode(ZkServerModeEnum.FOLLOWER);
        }
        zk.setServerStatus(ZkServerStatus.ONLINE);

        ZkServerInfo info = zsd.save(zk);
        return info;
    }

    @Override
    public List<Map<String, String>> getZkStatus() throws Exception {

        Map<String, String> allServerStat = new HashMap<>();
        List<Map<String,String>> res = new ArrayList<>();
        List<ZkServerInfo> zkNodes = zsd.findAll();
        for (ZkServerInfo zkNode : zkNodes) {
            allServerStat.put(zkNode.getIp(), zkNode.getServerStatus().getVal());
        }
        res.add(allServerStat);
        return res;
    }

    @Override
    public Map<String, List<String>> getZkStatus(String host, int port, String command) throws IOException {
        SocketClient socketClient = new SocketClient(host, port);
        Map<String,List<String>> hostAndRes = new HashMap<>();
        switch (command){
            case "stat":
                List<String> stats = socketClient.sendCommand(CommandConstEnum.STAT);
                hostAndRes.put(host,stats);
                return hostAndRes;
            case "conf":
                List<String> confs = socketClient.sendCommand(CommandConstEnum.CONF);
                hostAndRes.put(host,confs);
                return hostAndRes;
            case "cons":
                List<String> cons = socketClient.sendCommand(CommandConstEnum.CONS);
                hostAndRes.put(host,cons);
                return hostAndRes;
            case "crst":
                List<String> crst = socketClient.sendCommand(CommandConstEnum.CRST);
                hostAndRes.put(host,crst);
                return hostAndRes;
            case "dump":
                List<String> stat = socketClient.sendCommand(CommandConstEnum.STAT);
                if(stat.get(0).equals("leader")){
                    List<String> dump = socketClient.sendCommand(CommandConstEnum.DUMP);
                    hostAndRes.put(host,dump);
                    return hostAndRes;
                }else{
                    throw new RuntimeException("This zookeeper node not is leader.");
                }
            case "envi":
                List<String> envi = socketClient.sendCommand(CommandConstEnum.ENVI);
                hostAndRes.put(host,envi);
                return hostAndRes;
            case "ruok":
                List<String> ruok = socketClient.sendCommand(CommandConstEnum.RUOK);
                hostAndRes.put(host,ruok);
                return hostAndRes;
            case "srst":
                List<String> srst = socketClient.sendCommand(CommandConstEnum.SRST);
                hostAndRes.put(host,srst);
                return hostAndRes;
            case "srvr":
                List<String> srvr = socketClient.sendCommand(CommandConstEnum.SRVR);
                hostAndRes.put(host,srvr);
                return hostAndRes;
            case "wchs":
                List<String> wchs = socketClient.sendCommand(CommandConstEnum.WCHS);
                hostAndRes.put(host,wchs);
                return hostAndRes;
            case "wchp":
                List<String> wchp = socketClient.sendCommand(CommandConstEnum.WCHP);
                hostAndRes.put(host,wchp);
                return hostAndRes;
            case "wchc":
                List<String> wchc = socketClient.sendCommand(CommandConstEnum.WCHC);
                hostAndRes.put(host,wchc);
                return hostAndRes;
            default:
                throw new IllegalStateException("Unexpected value: " + command);
        }
    }

    @Override
    public List<ZkServerInfo> getZkNodeByClusterId(int clusterId) {
        return zsd.findAllByClusterId(clusterId);
    }

    @Override
    public String createZNode(String zkHostInfos, String path, String initData, CreateMode pathMode) throws Exception {
        ZookeeperClient client = new ZookeeperClient(zkHostInfos, 5000);
        boolean exists = client.exists(path);
        if(exists){
            throw new ZkException(path + " is already existed.");
        }
        String resPath = client.create(path, initData.getBytes(), pathMode);
        client.releaseConnection();
        return resPath;
    }

    @Override
    public void delZNode(String zkHostInfos, String path) throws Exception {
        ZookeeperClient client = new ZookeeperClient(zkHostInfos, 5000);
        boolean exists = client.exists(path);
        if(!exists){
            throw new ZkException(path + " is not existed.");
        }
        try {
            client.delete(path);
        }catch (Exception e){
            e.printStackTrace();
            throw new ZkException("delete " + path + "failed.\n");
        }finally {
            if(client != null){
                client.releaseConnection();
            }
        }
    }

    @Override
    public List<String> getZnodeChildren(String path, String hosts) throws Exception {
        ZookeeperClient client = new ZookeeperClient(hosts, 5000);
        List<String> children = client.getChildren(path);
        return children;
    }

    @Override
    public void updateStat(Integer zid, ZkServerModeEnum serverMode, ZkServerStatus stat) {
        zsd.updateZkStat(zid,serverMode,stat);
    }


    public Map<String,String> serverStatusByIpPort(String host,int port) throws IOException {
        Map<String,String> res = new HashMap<>();
        SocketClient socketClient = new SocketClient(host, port);
        String stat = socketClient.sendStat();
        res.put(host,stat);
        return res;
    }

    public Map<String,String> serversStatus(String hosts){
        List<String> ipPort = new ArrayList<>();
        Map<String,String> hostAndStat = new HashMap<>();
        String[] ipAndPort = hosts.split(",");

        for (String s : ipAndPort) {
            ipPort.add(s);
        }
        System.out.println(ipAndPort);

        return null;
    }

    public String getZkHostInfos(Integer cid) throws ZkException {
        StringBuffer buffer = new StringBuffer();
        List<ZkServerInfo> zkInfos = this.getZkNodeByClusterId(cid);

        if(zkInfos.size() == 0){
            throw new ZkException("Get zookeeper server info is empty.");
        }
        for (ZkServerInfo zkInfo : zkInfos) {
            buffer.append(zkInfo.getIp() + ":" + zkInfo.getPort() + ",");
        }

        return buffer.toString().substring(0,buffer.toString().length() - 1);
    }
}
